package com.liu.kafka.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;


/**
 * canal监听数据库数据同步处理
 *
 * @author liu
 */
@Slf4j
@Service
public class CanalConsumer {
    public static final String INSERT = "insert";
    public static final String UPDATE = "update";
    public static final String DELETE = "delete";

    /**
     * 监听数据库数据变化
     *
     * @param record
     * @param ack
     */
//    @KafkaListener(groupId = "data_sync", topics = {"canal_test"})
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.info("收到消息:{}", record.toString());
            handleRecord(record);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("数据同步失败:" + record.toString(), e);
            ack.nack(1000);
        }
    }

    private void handleRecord(ConsumerRecord<String, String> record) {
        CanalMessage canalMessage = JSON.parseObject(record.value(), CanalMessage.class);
        if (canalMessage == null || CollectionUtils.isEmpty(canalMessage.getData())) {
            log.info("canalMessage is null or data list is empty.");
            return;
        }

        String index = canalMessage.getDatabase() + "_" + canalMessage.getTable();

        //@TODO 同步数据
        switch (canalMessage.getType()) {
            case INSERT:


            case UPDATE:
                // 增加、修改处理
                for (JSONObject jsonObject : canalMessage.getData()) {

                    boolean success = true;
                    log.info("写入数据:{} 到 ES[index={}] {}", jsonObject.toJSONString(), index, success ? "成功" : "失败");
                }
                break;
            case DELETE:
                // 删除处理
                for (JSONObject jsonObject : canalMessage.getData()) {
                    boolean success = true;
                    log.info("删除数据:{} 从 ES[index={}] {}", jsonObject.toJSONString(), index, success ? "成功" : "失败");
                }
                break;
            default:
                break;
        }
    }
}
